package mq

import (
	"github.com/Shopify/sarama"
	"strings"
)

//var producer sarama.SyncProducer

var (
	KafkaConfig kafkaConfig
	producer    sarama.SyncProducer
	consumer    sarama.Consumer
	sendChan    chan *sarama.ProducerMessage
)

type kafkaConfig struct {
	Address string `ini:"address"`
}

func InitKafkaProducer() error {

	kafka := sarama.NewConfig()
	kafka.Producer.RequiredAcks = sarama.WaitForAll
	kafka.Producer.Partitioner = sarama.NewRandomPartitioner
	kafka.Producer.Return.Successes = true

	var err error
	producer, err = sarama.NewSyncProducer(strings.Split(KafkaConfig.Address, ","), kafka)
	if err != nil {
		return err
	}
	sendChan = make(chan *sarama.ProducerMessage, 10)

	//接收于监听
	go KafkaSend()

	return nil
}

func InitKafkaConsumer() error {
	var err error
	consumer, err = sarama.NewConsumer(strings.Split(KafkaConfig.Address, ","), nil)
	if err != nil {
		return err
	}
	return nil
}

func SendKafkaMsgChan(topic, text string) {
	msg := &sarama.ProducerMessage{
		Topic: topic,
		Value: sarama.StringEncoder(text),
	}
	sendChan <- msg
}
